package com.gopay.common.dss.lock.impl;

import com.gopay.common.constants.dic.GopayGlobalVariablesConstant;
import com.gopay.common.dss.DssException;
import com.gopay.common.dss.factory.CuratorObjectFactory;
import com.gopay.common.dss.lock.DssLock;
import com.gopay.common.dss.tools.FuseTool;
import com.gopay.common.util.spring.SpringContextUtil;
import org.apache.curator.framework.recipes.locks.InterProcessLock;
import org.apache.curator.framework.recipes.locks.InterProcessMultiLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * Created by zhaojt on 2017/7/27.
 * 基于zookeeper Curator框架的分布式锁实现
 * * 分布式可重入公平锁，在zookeeper 树形节点指定path下建立临时顺序子节点，获取锁顺序同申请锁顺序
 *  1.zookeeper在锁有效期内重启, 有序节点不会消失，重启后release释放锁无异常
 *  2.某JVM申请锁后与zookeeper断线，经过一段时间后zookeeper认为客户端可能宕机，将未到达有效期的节点删除，
 *    被误认为宕机的JVM重新连线后，release还是不抛异常，很坑。并且在该节点被zk删除后，其他线程申请锁能成功。
 *    此种极端情况，curator的分布式锁无法保证互斥性。
 *
 * 此对象非线程安全!
 *
 */
public class ZkDssLock implements DssLock {


    /**
     * Curator 在本地事务超时(事务执行时间>锁有效期)时，release方法不会抛异常。
     * 此功能需要应用端实现, 应用端记录事务超时时间
     */
    private long expireTime;

    private InterProcessLock mutex;

    /**
     * zookeeper的树形节点路径，会在改路径下建立临时顺序节点（排队等锁）
     */
    private List<String> paths;

    private Logger logger = LoggerFactory.getLogger(ZkDssLock.class);

    public ZkDssLock(List<String> paths) {
        CuratorObjectFactory factory = (CuratorObjectFactory)SpringContextUtil.getBean("curatorObjectFactory");
        long start = System.currentTimeMillis();
        this.paths = paths;

        this.mutex = new InterProcessMultiLock(factory.getCuratorFramework(), paths);

        logger.info("创建InterProcessMutex耗时 ; " + (System.currentTimeMillis() - start) + " 毫秒!");
    }

    @Override
    public boolean acquire(long time, TimeUnit unit) throws DssException {
        try {
            //计算过期时间  当前时间 + time
            expireTime = System.nanoTime() + unit.toNanos(time);
            logger.info("准备获取zookeeper分布式锁");
            long start = System.currentTimeMillis();
            boolean acquire = mutex.acquire(time, unit);
            logger.info("Acquire zkDssLock 耗时 ：" + (System.currentTimeMillis() - start) + " 毫秒! paths = " + paths);
            return acquire;//网络问题可能会导致长时间尝试连接zookeeper，生产环境参数设置
        } catch (Exception e) {
            logger.error("Acquire zookeeper DssLock Exception : " + e.getMessage());
            fuse(e);
            throw new DssException("调用zookeeper获取锁异常", e);
        }
    }

    @Override
    public void release() throws DssException {
        if(mutex != null && mutex.isAcquiredInThisProcess()){
            boolean isExpired = false;
            try {
                logger.info("准备释放zookeeper分布式锁");//网络异常会尝试释放，会阻塞很长一段时间，生产环境参数设置

                long start = System.currentTimeMillis();
                mutex.release();
                logger.info("Release zkDssLock 耗时 ：" + (System.currentTimeMillis() - start) + " 毫秒!");

                //当前时间 > 过期时间  当前事务超时不可信
                if(System.nanoTime() > expireTime){
                    isExpired = true;
                    throw new DssException("超时事务不可信!");
                }
            } catch (Exception e) {
                logger.error("Release zookeeper DssLock Exception : " + e.getMessage());
                //不是因当前事务超时引起的异常需要熔断
                //release 暂时不熔断
//                if(!isExpired){
//                    fuse(e);
//                }
                throw new DssException("调用zookeeper释放锁异常", e);
            }
        }
    }

    public void fuse(Exception e){
        FuseTool fuseTool = (FuseTool) SpringContextUtil.getBean("fuseTool");

        try {
            boolean isFuse = fuseTool.fuse(GopayGlobalVariablesConstant.ZK_LOCK_SWITCH,
                    GopayGlobalVariablesConstant.ZK_LOCK_FUSE_MOBILE, 5);

            if(isFuse){
                logger.info("准备关闭CuratorFramework...");
                CuratorObjectFactory factory = (CuratorObjectFactory)SpringContextUtil.getBean("curatorObjectFactory");
                factory.closeCuratorFramework();
            }
        }catch (Exception ex){
            logger.error("熔断zookeeper分布式锁开关出现异常," , ex);
        }

    }

    public List<String> getPaths() {
        return paths;
    }

    public void setPaths(List<String> paths) {
        this.paths = paths;
    }
}
